#include <config.h>
#include <core.h>


static void msgd_worker_process_cycle(void *data);
static msgd_int_t msgd_worker_process_init();
static void * thread_worker_cycle(void *arg);


static msgd_int_t msgd_trylock_accept_mutex();
static msgd_int_t msgd_enable_accept_events();
static msgd_int_t msgd_disable_accept_events();

msgd_shmtx_t   accept_mutex;
msgd_uint_t    msgd_use_accept_mutex = 1;
msgd_uint_t    msgd_accept_mutex_held = 0;
msgd_int_t     msgd_accept_disabled;

msgd_event_t         *msgd_posted_events;
//extern block_queue_t  msgd_posted_events_queue;

msgd_int_t msgd_init_process_environ()
{
  msgd_shm_t  accept_shm;

  msgd_pid  = getpid();
  msgd_ncpu = sysconf(_SC_NPROCESSORS_ONLN);
  INFO("msgd_ncpu:%d",msgd_ncpu);

  accept_mutex.spin = -1;

  accept_shm.size = sizeof(msgd_atomic_t)*2;
  if(msgd_shm_alloc(&accept_shm) == MSGD_ERROR){
     ERROR("msgd_shm_alloc error");  
     return MSGD_ERROR;
  }


  if(msgd_shmtx_create(&accept_mutex,(msgd_shmtx_sh_t*)accept_shm.addr) == MSGD_ERROR){
     ERROR("msgd_shmtx_create error");
     return MSGD_ERROR;
  }



  return MSGD_OK;
}



void msgd_start_worker_processes()
{
  msgd_uint_t i;
  //msgd_channel_t channel; 

  if(global.processes_n <= 1){

     msgd_use_accept_mutex = 0;
     msgd_worker_process_cycle(NULL);
 
  }else{

     for( i = 0; i < global.processes_n; i++){
       msgd_spawn_process(msgd_worker_process_cycle, NULL,
                          "worker process");  
     }

  }
}


void msgd_spawn_process(msgd_worker_process_cycle_pt proc,void *data,char *name)
{
  pid_t p;

  p = fork();

  switch(p){
     case -1:
       ERROR("msgd_spawn_process fork() failed");
       break;
     case 0: //child
       msgd_pid = getpid();
       proc(data);
       break;
     default:
       break;  
  }

  DEBUG("start %s %ld",name,p);
}



static void msgd_worker_process_cycle(void *data)
{
   INFO("msgd_worker_process_cycle data:%p",data); 

   set_proctitle((u_char*)"worker proc");
   (void)msgd_worker_process_init();

   for(;;){
      // if exit
      msgd_process_events_and_timers();
      // if terminate
      // if reopen
      // if quit
   }
}


void msgd_process_events_and_timers()
{
   msgd_int_t  timer = MSGD_TIMER_INFINITE;
   msgd_uint_t delta , flags = 0;

   timer = msgd_event_find_timer();
   flags = MSGD_PROCESS_EVENT_UPDATE_TIME;

   if(global.worker_threads_n > 0 && (timer == MSGD_TIMER_INFINITE || timer > 500) ){
      timer  = 500; //for expire timer  because other thread use msgd_add_timer
   }

   if(msgd_use_accept_mutex){
  
     if(msgd_accept_disabled > 0){
        msgd_accept_disabled--;
     }else if((global.worker_threads_n > 0) && ((msgd_uint_t)msgd_posted_events_queue.size >= global.worker_threads_n)){
        DEBUG("accept disabled");
     }else{

         if(msgd_trylock_accept_mutex() == MSGD_ERROR){
             return ;
         }

         if(msgd_accept_mutex_held){
            flags |= MSGD_PROCESS_EVENT_POST_EVENTS;
         }else{
            if(timer == MSGD_TIMER_INFINITE || timer > 500)
               timer = 500; 
         }
     }
   }
 
   delta = msgd_current_msec;

   (void)msgd_process_events(timer,flags);

   delta = msgd_current_msec - delta;

   INFO("delta %d",delta);

   if(msgd_accept_mutex_held)
      msgd_shmtx_unlock(&accept_mutex);

   if(delta)
     msgd_expire_timers();

   if(msgd_posted_events){
      msgd_handle_posted_event(&msgd_posted_events);
   }

   INFO("msgd_process_events_and_timers end",msgd_pid);
}


static msgd_int_t msgd_trylock_accept_mutex()
{
   if(msgd_shmtx_trylock(&accept_mutex)){
      DEBUG("accept mutex locked");
      if(msgd_accept_mutex_held){
         return MSGD_OK;
      }
      if(msgd_enable_accept_events() == MSGD_ERROR){
         msgd_shmtx_unlock(&accept_mutex);
         return MSGD_ERROR;
      }
      msgd_accept_mutex_held = 1;
      return MSGD_OK;
   }

   DEBUG("accept mutex lock failed");
   if(msgd_accept_mutex_held){
      if(msgd_disable_accept_events() == MSGD_ERROR){
         return MSGD_ERROR;
      }
      msgd_accept_mutex_held = 0;
   }

  
   return MSGD_OK;
}



static msgd_int_t msgd_enable_accept_events()
{
   msgd_listening_t *ls;
   msgd_connection_t *c;
   msgd_uint_t i;
   DEBUG("");
   for(i=0; i<global.listenings_n; i--){
      ls = &global.listenings[i]; 
      c  = ls->data;
      DEBUG("enable %s:%s accept %p",ls->addr_text,ls->port_text,c); 
      if(msgd_add_event(c->read,MSGD_READ_EVENT,EPOLLET) == MSGD_ERROR){
         ERROR("enable %s:%s accept failed",ls->addr_text,ls->port_text); 
         return MSGD_ERROR;
      } 
   } 
   
   DEBUG("enable accept end"); 
   return MSGD_OK;
}

static msgd_int_t msgd_disable_accept_events()
{
   msgd_listening_t *ls;
   msgd_connection_t *c;
   msgd_uint_t i;
   for(i=0; i<global.listenings_n; i--){
      ls = &global.listenings[i];
      c  = ls->data;
    
      if(!c->read->active){
        continue;
      }

      if(msgd_del_event(c->read,MSGD_READ_EVENT,MSGD_DISABLE_EVENT) == MSGD_ERROR){
         return MSGD_ERROR;
      } 
   } 
   
   return MSGD_OK;
}

static msgd_int_t msgd_worker_process_init()
{
   msgd_uint_t        n           = global.connections_n , tn = global.worker_threads_n;
   pool_t            *pool        = global.pool;
   msgd_connection_t **connections = &global.connections , *c , *next;
   msgd_connection_t **free_connections = &global.free_connections;
   msgd_event_t      **read_events =  &global.read_events  , *rev;
   msgd_event_t      **write_events = &global.write_events;
   pthread_t         **worker_threads = &global.worker_threads , *tid;

   INFO("init process  :%p  %d",&global,global.processes_n);

   block_queue_init(&msgd_posted_events_queue,0,1); // max size and blocked queue


   if(msgd_init_event() == MSGD_ERROR)
      return MSGD_ERROR;

   *connections = pool_palloc(pool, (sizeof(msgd_connection_t) * n));
   if(*connections == NULL){
      return MSGD_ERROR; 
   }

   *read_events = pool_palloc(pool,(sizeof(msgd_event_t) * n)); 
   if(*read_events == NULL)
     return MSGD_ERROR;

   *write_events = pool_palloc(pool,(sizeof(msgd_event_t) * n)); 
   if(*write_events == NULL)
     return MSGD_ERROR;

   if(tn > 0){
     *worker_threads = pool_palloc(pool,(sizeof(pthread_t) * tn));
     if(*worker_threads == NULL)
        return MSGD_ERROR;

     msgd_init_threads(tn,1024*1024);
     size_t i;
     for(i=0; i<tn; i++){
        tid = &(*worker_threads)[i];
        if(msgd_create_thread(tid,thread_worker_cycle,NULL) == MSGD_ERROR){
           return MSGD_ERROR; 
        }
     }
   }
 

   c = *connections;
   next = NULL; 

   do{
     n--;

     c[n].data =  next;
     c[n].read =  &(*read_events)[n];
     c[n].write = &(*write_events)[n];
     c[n].fd    = -1;
     c[n].close = 0;
     queue_init(&c[n].queue);

     next = &c[n];

   }while(n);

   *free_connections = next;
   global.free_connections_n = global.connections_n;
 
   assert(global.free_connections == global.connections);

   msgd_listening_t *ls;
   for(n=0; n<global.listenings_n; n--){
      ls = &global.listenings[n]; 
      c =  msgd_get_connection(ls->fd);
      if(c==NULL){
         return MSGD_ERROR;
      }

      c->listening = ls;
      ls->data = c;

      rev = c->read;
      
      rev->accept = 1; 
      rev->handler = msgd_event_accept; 

      if(msgd_use_accept_mutex)
         continue;

      if(msgd_add_event(rev,MSGD_READ_EVENT,EPOLLET) == MSGD_ERROR){
         return MSGD_ERROR;
      }
   }

   return MSGD_OK;
}

static void * thread_worker_cycle(void *arg)
{
   DEBUG("thread %d created arg:%p",pthread_self(),arg);

   for(;;){

      queue_t *qq  = block_queue_pop(&msgd_posted_events_queue);
      if(qq == NULL && msgd_posted_events_queue.block_flag == 0){
         DEBUG("thread %d exiting",pthread_self());
         break; // thread exit;
      }

      msgd_event_t *ev = queue_data(qq,msgd_event_t,queue);
      DEBUG("thread worker get event:%p and goto execute",ev);
 
      ev->handler(ev);

   }
  
   return (void*)NULL;
}
